package com.mindmax.wordmemory.mq;

import com.alibaba.fastjson2.JSON;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.mindmax.wordmemory.domain.DataVocabularyItem;
import com.mindmax.wordmemory.domain.DataVocabularySource;
import com.mindmax.wordmemory.mapper.DataVocabularyItemMapper;
import com.mindmax.wordmemory.mapper.DataVocabularySourceMapper;
import com.mindmax.wordmemory.vo.VocabularySource;
import com.ruoyi.common.crawler.WebSiteEnum;
import com.ruoyi.common.crawler.Word;
import com.ruoyi.common.crawler.WordItem;
import com.ruoyi.common.crawler.WordTypeEnum;
import com.ruoyi.common.enums.DataSourceType;
import com.ruoyi.framework.datasource.DynamicDataSourceContextHolder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import org.zeromq.ZMQ;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.io.IOException;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

@Slf4j
@Component
public class ZeroMqReceiver {

    @Resource
    private DataVocabularySourceMapper dataVocabularySourceMapper;

    @Resource
    private DataVocabularyItemMapper dataVocabularyItemMapper;
    ZMQ.Context context = ZMQ.context(1);
    ZMQ.Socket socket = context.socket(ZMQ.PULL);

    @PostConstruct
    public void init() {
        log.info("##################################  start receive message:");
        context = ZMQ.context(1);
        socket = context.socket(ZMQ.PULL);
        socket.bind("tcp://localhost:5555");
        new Thread(this::receiveMessage).start(); // 启动一个新线程来处理消息
    }
    public void receiveMessage() {
        DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.WORD_MEMORY.name());
        while (true) {

            byte[] message = socket.recv(0);
            log.info("##################################  start receive message:  ################");

            log.info("################      消费者开始处理数据   ################");

            String json       = new String(message);
            if (StringUtils.isEmpty(json)) {
                continue;
            }
            List<VocabularySource> vocabularySources = JSON.parseArray(json, VocabularySource.class);

            for (VocabularySource vocabularySource : vocabularySources) {
                String word   = vocabularySource.getWord();
                Long   bankId = vocabularySource.getBankId();
                Long   uid    = vocabularySource.getUid();
                Word   myWord = null;
                try {
                    log.info("process word - {}", word);
                    myWord = WebSiteEnum.Oxford.process(word);
                    log.info("myWord is - {} success", JSON.toJSONString(myWord));
                    if (myWord != null) {
                        List<WordItem> wordItems = myWord.getWordItems();
                        if (CollectionUtils.isEmpty(wordItems)) {
                            log.info("####### word - {} has no word wordItems", word);;
                        }
                        if (CollectionUtils.isNotEmpty(wordItems)) {
                            for (WordItem wordItem : wordItems) {
                                Date                 date                 = new Date();



                                List<WordItem.MeaningItem> meaningItems = wordItem.getMeaningItems();
                                if (CollectionUtils.isEmpty(meaningItems)) {
                                    log.info("####### word - {} has no meaning", word);
                                    continue;
                                }
                                for (WordItem.MeaningItem meaningItem : meaningItems) {
                                    DataVocabularyItem dataVocabularyItem = new DataVocabularyItem();
                                    dataVocabularyItem.setWord(word);
                                    dataVocabularyItem.setUsPhonetics(wordItem.getUsPhonetic());
                                    dataVocabularyItem.setUkPhonetics(wordItem.getUkPhonetic());
                                    dataVocabularyItem.setUsVoice(wordItem.getUsTouch());
                                    dataVocabularyItem.setUkVoice(wordItem.getUkTouch());
                                    dataVocabularyItem.setWordType(Optional.ofNullable(wordItem.getWordTypeEnum())
                                            .map(WordTypeEnum::getAlias).orElse(""));
                                    dataVocabularyItem.setUid(String.valueOf(uid));
                                    dataVocabularyItem.setCreateTime(date);
                                    dataVocabularyItem.setUpdateTime(date);
                                    List<String> synonyms = wordItem.getSynonyms();
                                    if (CollectionUtils.isEmpty(synonyms)) {
                                        dataVocabularyItem.setSynonyms("");

                                    } else {
                                        synonyms = synonyms.stream().limit(6).collect(Collectors.toList());
                                        String join = Joiner.on(" / ").skipNulls().join(synonyms);
                                        dataVocabularyItem.setSynonyms(join);
                                    }

                                    dataVocabularyItem.setMeaning(meaningItem.getMeaning());
                                    dataVocabularyItem.setChineseMeaning(meaningItem.getChineseMeaning());
                                    if (CollectionUtils.isEmpty(meaningItem.getPictures())) {
                                        dataVocabularyItem.setPicture("");
                                    } else {
                                        dataVocabularyItem.setPicture(meaningItem.getPictures().get(0));
                                    }
                                    dataVocabularyItem.setChosenExample("");
                                    List<String> examples = meaningItem.getExamples();
                                    List<String> exampleChineseTranslations = meaningItem.getExampleChineseTranslations();
                                    if (CollectionUtils.isNotEmpty(examples) && CollectionUtils.isNotEmpty(exampleChineseTranslations)) {
                                        StringBuilder sb = new StringBuilder();
                                        for (int i = 0; i < Math.min(examples.size(), 3); i++) {
                                            sb.append(examples.get(i)).append("=").append(exampleChineseTranslations.get(i)).append("&");
                                        }
                                        dataVocabularyItem.setExamples(sb.toString());
                                    }
                                    dataVocabularyItem.setBankId(bankId);

                                    dataVocabularyItemMapper.insertDataVocabularyItem(dataVocabularyItem);
                                    log.info("insert word item - {} success", word);
                                }


                            }
                        }
                    }
                } catch (Exception e) {
                    log.error("process word - {} error", word, e);
                }
            }
            log.info("##################################  end receive message:");

        }

    }


    @PreDestroy
    public void close() {
        if (socket != null) {
            socket.close();
        }
        if (context != null) {
            context.term();
        }
    }
}
